#include "CQMyServer.h"

MyServer::MyServer(ip_port_s* ipc)
{
	std::thread th(std::bind(&MyServer::write_mq_task, this));
	th.detach();

	ips_.ip = ipc->ip;
	ips_.port = ipc->port;
	cbs_ = { on_read,on_connect,on_disconnect };
	server_ = CQNet()->getServer_Tcp(ips_.port, ips_.ip, &cbs_, this);
}

MyServer::MyServer(ip_port_s* ipc, io_service& io)
{
	std::thread th(std::bind(&MyServer::write_mq_task, this));
	th.detach();

	ips_.ip = ipc->ip;
	ips_.port = ipc->port;
	cbs_ = { on_read,on_connect,on_disconnect };
	server_ = CQNet()->getServer_Tcp(ips_.port, io, ips_.ip, &cbs_, this);
}

template<typename T>
void clear_queue(std::queue<T>& t)
{
	std::queue<T> empty;
	std::swap(empty, t);
}

MyServer::~MyServer()
{
	close_all();
	{
		std::lock_guard<std::mutex> lk(writeMU_);
		clear_queue(writeMQ_);
	}
	{
		std::lock_guard<std::mutex> lk(readMU_);
		clear_queue(readMQ_);
	}
}

void MyServer::on_read(int key, const char* msg, size_t len, void* ptr)
{
	MyServer* ms = (MyServer*)ptr;
	if (!ms) { logs(error_, funcx + "this ptr is null"); return; }
	std::string str = string(msg, len);
	logs(debug_, funcx + str);
	{
		std::unique_lock<std::mutex> lk(ms->readMU_);
		ms->readMQ_.push(ms->parse_read_msg(str));
		ms->readCV_.notify_one();
	}
}

void MyServer::on_connect(int key, void* ptr)
{
	MyServer* ser = (MyServer*)ptr;
	ser->client_size++;
	logs(debug_, funcx);
}

void MyServer::on_disconnect(int key, int error_code, void* ptr)
{
	MyServer* ser = (MyServer*)ptr;
	ser->client_size--;
	logs(debug_, funcx);
}

void MyServer::write_mq(CQ_MSG& cm)
{
	std::unique_lock<std::mutex> lk(writeMU_);
	writeMQ_.push(build_write_msg(cm));
	writeCV_.notify_one();
}

void MyServer::close_all()
{
	server_->close_all();
}

void MyServer::close_one(int key)
{
	server_->close_one(key);
}

void MyServer::stop_io()
{
	server_->stop_io_accept();
}

void MyServer::write_mq_task()
{
	while (!is_close_.load())
	{
		if (client_size <= 0) { Sleep(50); continue; }
		std::string msg = "";
		{
			std::unique_lock<std::mutex> lk(writeMU_);
			if (is_close_.load())return;
			writeCV_.wait(lk, [this] { return !writeMQ_.empty(); });
			msg = writeMQ_.front();
			writeMQ_.pop();
		}
		CQ_MSG cm = parse_read_msg(msg);
		if (cm.key_to != SEND_ALL::value)
			server_->write_to_one(cm.key_to, msg);
		else server_->write_to_all(msg);
	}
}

CQ_MSG MyServer::read()
{
	std::unique_lock<std::mutex> lk(readMU_);
	readCV_.wait(lk, [this] { return !readMQ_.empty(); });
	CQ_MSG msg = readMQ_.front();
	readMQ_.pop();

	return msg;
}